package simple;

import bean.Event;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.CloseableIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class ExampleCountSourceMain {
    private static final Logger logger = LoggerFactory.getLogger(ExampleCountSourceMain.class);

    public static void main(String[] args) throws Exception {
        //main1();
        logger.info("---------------------------------------");
        //main2();
        logger.info("---------------------------------------");
        main3();
    }


    public static void main1() throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        // 设置并行度，默认值 = 当前计算机的CPU逻辑核数（设置成1即单线程处理）
        //env.setMaxParallelism(1);
        //env.setParallelism(1);

        DataStream<Long> inputDataStream = env.addSource(new ExampleCountSource());

        // 基于数据流进行转换计算
        var resultStream = inputDataStream.map(t -> t + 100)
                .filter(t -> t > 1000).executeAndCollect();

        resultStream.forEachRemaining(t -> {
            logger.info("-----------------------" + t);
        });
    }

    public static void main2() throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        // 设置并行度，默认值 = 当前计算机的CPU逻辑核数（设置成1即单线程处理）
        //env.setMaxParallelism(1);
        //env.setParallelism(1);

        var stream = env.fromCollection(List.of(1,2,3,4,5,6));
        var a = stream.filter(t -> t > 5).executeAndCollect();
        stream.print();
        logger.info("------------------------------" + a);
    }

    public static void main3() throws Exception {
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        // 设置并行度，默认值 = 当前计算机的CPU逻辑核数（设置成1即单线程处理）
        //env.setMaxParallelism(1);
        env.setParallelism(1);

        var stream = env.fromCollection(List.of(
            new Event("ab","ss1", 1L),
            new Event("cd","ss2", 2L),
            new Event("ef","ss3", 3L)
        ));

        CloseableIterator<Tuple3<String, String, Long>> a = stream
                .map( event -> Tuple3.of(event.userId, event.eventType, event.timestamp))

                .returns(Types.TUPLE(Types.STRING,Types.STRING, Types.LONG))
                .executeAndCollect()
                ;  // 声明返回类型
        //stream3.print();
        logger.info("------------------------------" + a);
    }

}
